package cn.lecosa.spark.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;

@SuppressWarnings("deprecation")
public class HbaseClient {
	private static HBaseAdmin admin = null;
	private static Configuration config = null;
	static {
		try {
			config = new Configuration();
			config.set("hbase.zookeeper.quorum", "cdh2,cdh3,cdh4"); //2181 这里可以省略掉
			// conf.set("hbase.zookeeper.quorum", "192.168.1.10:2181");
			config.set("hbase.rootdir", "hdfs://192.168.1.10:9000/hbase");
			config.set("hbase.client.scanner.timeout.period", "5000000"); 
			if(admin==null) {
				admin = new HBaseAdmin(config);
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	
	public static synchronized HBaseAdmin getHBaseAdmin() {
		return admin;
	}
	
	public static synchronized Configuration getHBaseConfiguration() {
		return config;
	}
	
	
	public static void main(String[] args) throws Exception {
		// 创建表
		// hBase.createTable("stu","cf");
		// 遍历表
		getAllTable();
		// 插入一条数据
		// hBase.addOneRecord("stu","key1","cf","name","baozi");
		// hBase.addOneRecord("stu","key1","cf","age","23");
		// 得到一条数据
		// hBase.getKey("stu","key1");
		// 得到所有数据
		// hBase.getAllData("stu");
		// 删除一条记录
		// hBase.deleteOneRecord("stu","key1");
		// 删除表
		// hBase.deleteTable("stu");
		// scan过滤器的使用
		// hBase.getScanData("stu","cf","name");
		// rowFilter的使用
		// hBase.getRowFilter("waln_log","^*_201303131459\\d*$");

	}

	/**
	 * rowFilter使用
	 * 
	 * @param tableName
	 * @param reg
	 * @throws Exception
	 */
	private void getRowFilter(String tableName, String reg) throws Exception {
		@SuppressWarnings("resource")
		HTable hTable = new HTable(config, tableName);
		Scan scan = new Scan();
		RowFilter rowFilter = new RowFilter(CompareOp.EQUAL, new RegexStringComparator(reg));
		scan.setFilter(rowFilter);
		ResultScanner scanner = hTable.getScanner(scan);
		for (Result result : scanner) {
			System.out.println(new String(result.getRow()));
		}
	}

	/**
	 * scan过滤器
	 * 
	 * @param tableName
	 * @param family
	 * @param qualifier
	 * @throws Exception
	 */
	private void getScanData(String tableName, String family, String qualifier) throws Exception {
		@SuppressWarnings("resource")
		HTable hTable = new HTable(config, tableName);
		Scan scan = new Scan();
		scan.addColumn(family.getBytes(), qualifier.getBytes());
		ResultScanner scanner = hTable.getScanner(scan);
		for (Result result : scanner) {
			if (result.raw().length == 0) {
				System.out.println(tableName + " 表数据为空！");
			} else {
				for (KeyValue kv : result.raw()) {
					System.out.println(new String(kv.getKey()) + "\t" + new String(kv.getValue()));
				}
			}
		}
	}

	/**
	 * 删除表
	 * 
	 * @param tableName
	 */
	private void deleteTable(String tableName) {
		try {
			if (admin.tableExists(tableName)) {
				admin.disableTable(tableName);
				admin.deleteTable(tableName);
				System.out.println("表:" + tableName + " 删除成功");
			}
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println("表" + tableName + "删除失败");
		}
	}

	/**
	 * 删除一条记录
	 * 
	 * @param tableName
	 * @param rowKey
	 */
	public void deleteOneRecord(String tableName, String rowKey) {
		@SuppressWarnings("resource")
		HTablePool hTablePool = new HTablePool(config, 1000);
		HTableInterface table = hTablePool.getTable(tableName);
		Delete delete = new Delete(rowKey.getBytes());
		try {
			table.delete(delete);
			System.out.println(rowKey + "记录删除成功！");
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println(rowKey + "记录删除失败！");
		}
	}

	/**
	 * 得到全部数据
	 * 
	 * @param string
	 * @throws Exception
	 */
	private void getAllData(String tableName) throws Exception {
		@SuppressWarnings("resource")
		HTable hTable = new HTable(config, tableName);
		Scan scan = new Scan();
		ResultScanner scanner = hTable.getScanner(scan);
		for (Result result : scanner) {
			if (result.raw().length == 0) {
				System.out.println(tableName + "  为空");
			} else {
				for (KeyValue kvKeyValue : result.raw()) {
					System.out.println(new String(kvKeyValue.getKey()) + "\t" + new String(kvKeyValue.getValue()));
				}
			}
		}
	}

	/**
	 * 得到一条数据
	 * 
	 * @param string
	 * @param string2
	 */
	private void getKey(String tableName, String rowkey) {
		@SuppressWarnings("resource")
		HTablePool hTablePool = new HTablePool(config, 1000);
		HTableInterface table = hTablePool.getTable(tableName);
		Get get = new Get(rowkey.getBytes());
		try {
			Result result = table.get(get);
			if (result.raw().length == 0) {
				System.out.println("行键" + rowkey + "内容为空！");
			} else {
				for (KeyValue kv : result.raw()) {
					System.out.println(new String(kv.getKey()) + "\t" + new String(kv.getValue()));
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * 插入一条数据
	 * 
	 * @param string
	 * @param string2
	 * @param string3
	 * @param string4
	 * @param string5
	 */
	private void addOneRecord(String tableName, String rowkey, String column, String qua, String value) {
		@SuppressWarnings({ "resource" })
		HTablePool hTablePool = new HTablePool(config, 1000);
		HTableInterface table = hTablePool.getTable(tableName);
		Put put = new Put(rowkey.getBytes());
		put.add(column.getBytes(), qua.getBytes(), value.getBytes());
		try {
			table.put(put);
			System.out.println("行键" + rowkey + " 内容插入成功！");
		} catch (IOException e) {
			e.printStackTrace();
			System.out.println("行键 " + rowkey + " 插入内容失败！");
		}
	}

	/**
	 * 遍历表
	 * 
	 * @return
	 * @throws Exception
	 */
	public static List<String> getAllTable() throws Exception {
		ArrayList<String> tables = new ArrayList<String>();
		if (admin != null) {
			HTableDescriptor[] listTables = admin.listTables();
			if (listTables.length > 0) {
				for (HTableDescriptor tableDesc : listTables) {
					tables.add(tableDesc.getNameAsString());
					System.out.println(tableDesc.getNameAsString());
				}
			}
		}
		return tables;
	}
	// /hbase/data/default/stu

	/**
	 * 创建表
	 * 
	 * @param tableName
	 * @param column
	 * @throws Exception
	 */
	private void createTable(String tableName, String column) throws Exception {
		if (admin.tableExists(tableName)) {
			System.out.println("表已经存在！");
		} else {
			HTableDescriptor tableDesc = new HTableDescriptor(tableName);
			tableDesc.addFamily(new HColumnDescriptor(column.getBytes()));
			admin.createTable(tableDesc);
			System.out.println(tableName + "创建成功！");
		}
	}

}